package org.itew.mymapreduce.core.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import org.itew.mymapreduce.core.DataCollection;
import org.itew.mymapreduce.core.Partition;

public class SortedPartition<K,V> implements Partition<K,V> {


	@Override
	public Collection<Set<K>> partition(DataCollection<K,V> dataCollection, int reducerNumber) {
		
		//合并key的值
		Set<K> hashKeySet = dataCollection.keySet();
		
		SortedSet<K> sortedSet = new TreeSet<K>();
		
		for(K key:hashKeySet){
			sortedSet.add(key);
		}
		
		//创建分任务
		List<Set<K>> partedKeySet = new ArrayList<Set<K>>(reducerNumber);
		
		if(reducerNumber==1){
			 partedKeySet.add(sortedSet);
			 return partedKeySet;
		}
		
		for(int i=0;i<reducerNumber;i++){
			partedKeySet.add(new TreeSet<K>());
		}
		
		int i = 0;
		int size = sortedSet.size();
		for(K key:sortedSet){
			partedKeySet.get((i*reducerNumber)/size).add(key);
			i++;
		}
		
		return partedKeySet;
	}

}
